package com.zhang.spark_2.com.zhang.core.req

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @title: 热门商品Top10
 * @author: zhang
 * @date: 2022/2/16 17:16 
 */
object Spark01_req_pro {

  def main(args: Array[String]): Unit = {
    //TODO 获取环境
    val conf = new SparkConf().setMaster("local").setAppName("WordCount")
    val sc = new SparkContext(conf);
    val userVisit: RDD[String] = sc.textFile("data/user_visit_action.txt")
    userVisit.cache()
    // (品类，点击数量）

    val clickCnt: RDD[(String, Int)] = userVisit
      .filter(
        line => {
          val datas: Array[String] = line.split("_")
          datas(6) != "-1"
        }
      )
      .map(
        line => {
          val datas: Array[String] = line.split("_")
          (datas(6), 1)
        }
      )
      .reduceByKey(_ + _)

    //统计下单的点击数量：（品类ID，下单数量）
    val orderCnt: RDD[(String, Int)] = userVisit
      .filter(
        line => {
          val datas: Array[String] = line.split("_")
          datas(8) != "null"
        }
      )
      .flatMap(
        line => {
          val datas: Array[String] = line.split("_")
          val cids: Array[String] = datas(8).split(",")
          cids.map((_, 1))
        }
      )
      .reduceByKey(_ + _)

    //统计支付的点击数量：（品类ID，支付数量）
    val payCnt: RDD[(String, Int)] = userVisit
      .filter(
        line => {
          val datas: Array[String] = line.split("_")
          datas(10) != "null"
        }
      )
      .flatMap(
        line => {
          val datas: Array[String] = line.split("_")
          val cids: Array[String] = datas(10).split(",")
          cids.map((_, 1))
        }
      )
      .reduceByKey(_ + _)

    // todo 转换结构
    val click: RDD[(String, (Int, Int, Int))] = clickCnt.mapValues((_, 0, 0))
    val order: RDD[(String, (Int, Int, Int))] = orderCnt.mapValues((0, _ ,0))
    val pay: RDD[(String, (Int, Int, Int))] = clickCnt.mapValues((0, 0,_))
    click.union(order).union(pay).reduceByKey{
      case (t1,t2)=>{
        (t1._1+t2._1,t1._2+t2._2,t1._3+t2._3)
      }
    }.sortBy(_._2,false).take(10).foreach(println)


    sc.stop()
  }
}
